package com.rem.flink.flink10Sql;

import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 追加查询  将一段持续查询的结果追加到结果表中 只有insert的操作 可以直接调用toDataStream() 操作
 *
 * @author Rem
 * @date 2022-11-07
 */

public class AppendQueryTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取数据源，并分配时间戳、生成水位线
        SingleOutputStreamOperator<Event> eventStream = env
                .fromElements(
                        new Event("Alice", "./home", 1000L),
                        new Event("Bob", "./cart", 1000L),
                        new Event("Alice", "./prod?id=1", 25 * 60 * 1000L),
                        new Event("Alice", "./prod?id=4", 55 * 60 * 1000L),
                        new Event("Bob", "./prod?id=5", 3600 * 1000L + 60 * 1000L),
                        new Event("Cary", "./home", 3600 * 1000L + 30 * 60 * 1000L),
                        new Event("Cary", "./prod?id=7", 3600 * 1000L + 59 * 60 * 1000L)
                )
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()));


        //创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //将数据流转换成指定的表，并且指定属性 指定 时间戳只支持timestamp 类型 不支持Long类型
        /**
         * 方法过期
         * 将数据流转换成表，并指定时间属性
         */
        Table eventTable = tableEnv.fromDataStream(
                eventStream,
                $("user"),
                $("url"),
                $("timestamp").rowtime().as("ts"));
        eventTable.printSchema();

        /**
         * 新版本
         * 老版本 .rowtime()  对应水位线 watermark
         */
        Schema build = Schema.newBuilder()
                .column("user", DataTypes.STRING())
                .column("url", DataTypes.STRING())
                .columnByExpression("ts", "CAST(TO_TIMESTAMP(FROM_UNIXTIME(`timestamp`)) AS TIMESTAMP(3))")
                .watermark("ts", "SOURCE_WATERMARK()")
                .build();
        Table eventTable2 = tableEnv.fromDataStream(eventStream, build);
        /**
         * (
         *   `user` STRING,
         *   `url` STRING,
         *   `ts` TIMESTAMP(3) *ROWTIME*
         * )
         */
        eventTable2.printSchema();

        //在环境中注册虚拟表
        tableEnv.createTemporaryView("eventTable", eventTable2);
        tableEnv.toDataStream(eventTable2).print("eventTable2:");

        // 设置累积窗口，执行SQL统计查询
        Table result = tableEnv
                .sqlQuery("SELECT " +
                        "user, " +
                        "window_end AS endT, " +            // 窗口结束时间
                        "COUNT(url) AS cnt " +              // 统计 url 访问次数
                        "FROM TABLE( " +
                        "CUMULATE( TABLE eventTable, " +    // 定义累积窗口
                        " DESCRIPTOR(ts), " +
                        "INTERVAL '30' MINUTE, " +          //固定表达式 30分钟
                        "INTERVAL '1' HOUR)) " +
                        "GROUP BY user, window_start, window_end "
                );

        tableEnv.toDataStream(result).print("result:");
        env.execute();
    }
}
